#include "SearchControllerThread.h"
#include "ISearchControllerThreadCallback.h"

#include "Core/Commands.h"
#include "Core/Message.h"

#include "Config/IConfig.h"

#include <QTimer>

#include <boost/lexical_cast.hpp>


namespace {
    const int g_freeThreadState = 0;
    const int g_busyThreadState = 1;

    /** If threre is no available service slot, try again after specified timeout */
    const unsigned long g_repeateRequestTimoutMsec = 500;

}

SearchControllerThread::SearchControllerThread(QObject *parent, IConfig *config, ISearchControllerThreadCallback *callback)
    : QThread(parent)
    , m_busyState(g_freeThreadState)
    , m_stopRequested(false)
    , m_callback(callback)
    , m_config(config)
    , m_boxManager(NULL)
{
    Core::BoxApi::ApiManagerParams boxManagerParams;
    boxManagerParams.conccurency = 1;
    boxManagerParams.connectionCloneAtDone = false;

    m_boxManager.reset(new Core::BoxApi::ApiManager(boxManagerParams));
}

SearchControllerThread::~SearchControllerThread()
{
    interrup();
}

bool SearchControllerThread::pushTask(SearchTaskPtr task)
{
    if (!m_busyState.testAndSetRelaxed(g_freeThreadState, g_busyThreadState))
        return false;

    m_mutex.lock();
        m_activeTask = task;
        m_condition.wakeOne();
    m_mutex.unlock();

    return true;
}

void SearchControllerThread::interrup()
{

    m_mutex.lock();
        boost::shared_ptr<Core::EndUserAPI::ServiceConnector> ptr = m_service;
        m_stopRequested = true;
        m_condition.wakeOne();
    m_mutex.unlock();

    try
    {
        m_boxManager->interrupt();
    }
    catch (...)
    {

    }

    QThread::wait();
}

void SearchControllerThread::run()
{
    while (true)
    {
        m_mutex.lock();
            if (m_stopRequested)
            {
                m_mutex.unlock();
                break;
            }

            if (NULL == m_activeTask.get())
                m_condition.wait(&m_mutex);

            if (m_stopRequested)
            {
                m_mutex.unlock();
                break;
            }
        m_mutex.unlock();

        processNextTask();

        m_activeTask = SearchTaskPtr();
        m_busyState = g_freeThreadState;
    }
}


void SearchControllerThread::processNextTask()
{
    const int serviceCount = m_config->getSlotsCount();
    if (0 == serviceCount)
    {
        m_callback->onError(m_activeTask, this, "Sevices list is empty");
        return;
    }

    /** Request accept timeout */
    const unsigned long g_requestProcessTimoutMsec = 180000; // 3 min

    QTimer timer;
    timer.setSingleShot(true);
    timer.start(g_requestProcessTimoutMsec);

    Core::BoxApi::BoxConnectionParams connectionParams;
    boost::shared_ptr<Core::BoxApi::IBoxConnection> boxConnection;

    ServiceSlot serviceSlot;
    while (true)
    {
        for (int i = 0; i < serviceCount; ++i)
        {
            if (!m_config->getSlotByIndex(i, serviceSlot))
                continue;

            connectionParams.host = serviceSlot.mHost;
            connectionParams.port = boost::lexical_cast<std::string>(serviceSlot.mPort);
            connectionParams.name = "ServiceConnection";
            connectionParams.id = i;

            try
            {
                boxConnection = m_boxManager->connect(connectionParams);
            }
            catch(std::exception & exc)
            {
                // TODO: may be send  message to log
                continue;
            }

            if (!m_stopRequested)
            {
                const TaskStatus status  = tryService(boxConnection);
                boxConnection->close();

                if (ServiceOutOfOrder == status)
                    continue;
            }

            boxConnection->close();

            return;
        }

        // sleep before again attempt
        m_mutex.lock();
        m_condition.wait(&m_mutex, g_repeateRequestTimoutMsec);
        m_mutex.unlock();

        if (m_stopRequested)
        {
            m_callback->onInterrupted(m_activeTask, this);
            return;
        }

        // If find service opertiontimeout elapsed, stop task processing
        if (!timer.isActive())
        {
            m_callback->onError(m_activeTask, this, "service is unavailable for a long time. timeout");
            return;
        }
    }
}

SearchControllerThread::TaskStatus SearchControllerThread::tryService(Core::BoxApi::IBoxConnectionPtr serviceConnector)
{
/*    // Firt send ping command

    Core::ICommandPtr pingCommand(new Core::PingCommand());

    try
    {
        Core::ICommandPtr reply(serviceConnector->sendMessageSync(pingCommand));

        // check if reply is OK
        if (Core::Aux::OK_TAG != reply->getTag())
            return ServiceOutOfOrder;
    }
    catch (std::exception & exc)
    {
            return ServiceOutOfOrder;
    }*/


    try
    {
        Core::ICommandPtr reply;
        // send search message
        if (ESearchTypeFile == m_activeTask->getType())
        {
            Core::ICommandPtr scanCommand(new Core::FileScanCommand(m_activeTask->getFilename().c_str()));

            reply = serviceConnector->sendMessageSync(scanCommand);
        }
        else
        {
            SearchTask::ByteArray & byteArray = m_activeTask->getByteArray();
            const std::size_t totalSize = byteArray.size();
            const std::size_t chunkSize = 1024;

            Core::ICommandPtr scanCommand(new Core::ArrayScanCommand(chunkSize, totalSize));

            m_boxManager->postMessage(serviceConnector, scanCommand);

            std::size_t chunksCount = totalSize / chunkSize;

            if (0 != totalSize % chunkSize)
                ++chunksCount;

            std::vector<unsigned char> chunkData;
            for (int i = 0; i < chunksCount-1; ++i)
            {
                chunkData.assign((unsigned char *)&byteArray[i * chunkSize], (unsigned char *)&byteArray[(i + 1) * chunkSize]);
                const std::size_t sentBytes = serviceConnector->writeRawData(chunkData);
                if (sentBytes != chunkSize)
                    throw std::runtime_error("socket write error");
            }

            std::size_t chunkTailSize = totalSize - (chunksCount * chunkSize);
            if (0 == chunkTailSize)
                chunkTailSize = chunkSize;

            unsigned char * startPtr = (unsigned char *)&byteArray[(chunksCount - 1) * chunkSize];
            chunkData.assign(startPtr, startPtr + chunkTailSize);

            const std::size_t sentBytes = serviceConnector->writeRawData(chunkData);
            if (sentBytes != chunkTailSize)
                throw std::runtime_error("socket write error");

            boost::tuple<Core::ICommandPtr, Core::BoxApi::IBoxConnectionPtr> nextMsg(m_boxManager->getNextMessage());
            reply = nextMsg.get<0>();

        }

        const int tag = reply->getTag();
        if (Core::Aux::GUID_TAG == tag)
        {
            Core::GuidReplyCommandPtr guidFoundReply = Core::commandCast<Core::GuidReplyCommand>(reply);
            const std::string & guid = guidFoundReply->getGuid();
            if (guid.empty())
            {
                m_callback->guidFound(m_activeTask, this, guid);
            }
            else
            {
                m_callback->notFound(m_activeTask, this);
            }
            return ServiceOutOfOrder;
        }
        else if (Core::Aux::ERROR_TAG == tag)
        {
            Core::ErrorCommandPtr errorCommand = Core::commandCast<Core::ErrorCommand>(reply);
            switch (errorCommand->getErrorCode())
            {
            case Core::ErrorCommand::INTERNAL_SERVER_ERROR:
                return ServiceOutOfOrder;
            case Core::ErrorCommand::FILE_NO_FOUND:
                m_callback->onError(m_activeTask, this, "File not found");
                return TaskCompleted;
            case Core::ErrorCommand::CONNECTION_BROKEN:
                return ServiceOutOfOrder;
            case Core::ErrorCommand::INTERNAL_PROTOCOL_ERROR:
                m_callback->onError(m_activeTask, this, errorCommand->getMessage());
                return TaskCompleted;
            default:
                // unknows error code. gets message
                m_callback->onError(m_activeTask, this, errorCommand->getMessage());
                return TaskCompleted;
            }
        }
        else
        {
            // undefined server reply
            m_callback->onError(m_activeTask, this, "server protocol error");
            return TaskCompleted;
        }

    }
    catch (std::exception & exc)
    {
        return ServiceOutOfOrder;
    }

}


